package com.example.KafkaDdemo.filter;

import com.example.KafkaDdemo.uitls.BytesUtils;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class ConsumerinterceptorTTL implements ConsumerInterceptor<String, String> {

    private static final long EXPIRE_INTERVAL = 10 * 1000;

    /**
     * 书本笔记：
     * 该方法在poll方法返回之前调用。调用结束后poll方法就返回消息了。
     * 该方法可以修改消费者消息，返回新的消息。拦截器可以过滤收到的消息或生成新的消息。
     *
     * 自己笔记：这个方法会先于KafkaConsumer中的 @KafkaListener 修饰的方法，这个过滤器就是针对这个方法前后做的aop
     * 自己理解。。。。
     *
     * 存在的问题：
     *  可能导致超时消息不能提交
     *  比如我消费者批量拉取10个数据，最后一个数据是超时消息，并且判断出来超时了，就不能被消费者消费切提交，前面9个消费完会自动提交
     */
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {

        long now = System.currentTimeMillis();
        // 用来存放过滤后的新消息
        Map<TopicPartition, List<ConsumerRecord<String, String>>> newRecords = new HashMap<>();

        for (TopicPartition tp : records.partitions()) {
            List<ConsumerRecord<String, String>> tpRecords =
                    records.records(tp);

            List<ConsumerRecord<String, String>> newTpRecords = new ArrayList<>();
            for (ConsumerRecord<String, String> record : tpRecords) {
                Headers headers = record.headers();
                long ttl = -1;
                for (Header header : headers) {
                    if (header.key().equalsIgnoreCase("ttl")){
                        ttl = BytesUtils.bytesToLong(header.value());
                    }
                }
                // ttl>0用来判断是否为延迟消息
                // now - record.timestamp() < ttl*1000  判断这个消息是否超时
                // 正常情况，就是这个消息要在5s内消费，但是如果判断出 现在时间-消息时间 < 5s 说明这个消息没有超时
                // 如果大于5s,说明超时了，这里的逻辑是丢弃
                if (ttl>0 && now - record.timestamp() < ttl*1000){
                    newTpRecords.add(record);
                }else if (ttl < 0){
                    newTpRecords.add(record);
                }

            }
            if (!newTpRecords.isEmpty()) {
                newRecords.put(tp, newTpRecords);
            }
        }
        return new ConsumerRecords<>(newRecords);

    }

    /**
     * 当消费者提交偏移量时，调用该方法。通常你可以在该方法中做一些记账类的动作，比如打日志等。
     * 调用者将忽略此方法抛出的任何异常。
     */
    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        offsets.forEach((tp, offset) -> System.out.println(tp + ":" + offset.offset()));
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
